package com.ccp.dev.im.server.connertor.impl;

import com.ccp.dev.im.model.ImUser;
import com.ccp.dev.im.server.connertor.ImConnertor;
import com.ccp.dev.im.server.constant.Constants;
import com.ccp.dev.im.server.group.ImChannelGroup;
import com.ccp.dev.im.server.model.MessageWrapper;
import com.ccp.dev.im.server.model.Session;
import com.ccp.dev.im.server.model.proto.MessageProto;
import com.ccp.dev.im.server.proxy.MessageProxy;
import com.ccp.dev.im.server.session.impl.SessionManagerImpl;
import com.ccp.dev.im.service.ImRoomUserService;
import com.ccp.dev.im.service.ImUserService;
import com.ccp.dev.system.model.SysUser;
import com.ccp.dev.system.service.SysUserService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.session.SessionInformation;
import org.springframework.security.core.session.SessionRegistry;
import org.springframework.security.core.userdetails.UserDetails;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

@Component
public class ImConnertorImpl implements ImConnertor {
    private final static Logger log = LoggerFactory.getLogger(ImConnertorImpl.class);
    @Resource
    private SessionManagerImpl sessionManagerImpl;

    @Resource
    private ImRoomUserService imRoomUserService;

    @Resource
    private MessageProxy messageProxyImpl;

    @Resource
    private SysUserService sysUserService;

    @Resource
    private ImUserService imUserService;

    @Resource
    private SessionRegistry sessionRegistry;

    @Override
    public void heartbeatToClient(ChannelHandlerContext hander, MessageWrapper wrapper) {
        //设置心跳响应时间
        hander.channel().attr(Constants.SessionConfig.SERVER_SESSION_HEARBEAT).set(System.currentTimeMillis());
    }

    @Override
    public void pushGroupMessage(MessageWrapper wrapper)
            throws RuntimeException {
        //这里判断群组ID 是否存在 并且该用户是否在群组内
        MessageProto.Model msg = (MessageProto.Model) wrapper.getBody();
        String roomId = msg.getGroupId();
        ChannelGroup groups = ImChannelGroup.CHANNELGROUP;
        Session[] sessions = sessionManagerImpl.getSessions();
        Set<String> userIdSet = imRoomUserService.getRoomUserIdsByRoomId(roomId);
        //查询发送消息的用户是否在这个群聊内
        String senderId = msg.getSender();
		boolean isExistOneRoom = imRoomUserService.qryUserExistOneRoom(senderId, roomId);
        for (Channel channel : groups) {
            for (Session session : sessions) {
                if (session.getNid().equals(channel.id().toString())) {
                    if (userIdSet.contains(session.getUserId())) {
                        if (isExistOneRoom) {
                            channel.attr(Constants.SessionConfig.SERVER_SESSION_HEARBEAT).set(System.currentTimeMillis());
                            ImChannelGroup.broadcastGroup(channel, wrapper.getBody());
                        }
                    }
                }
            }
        }
        if (isExistOneRoom) {
            messageProxyImpl.saveMessageToDB(wrapper);
        }
    }

    @Override
    public void pushSystemMessage(String obj, int msgType, String... userIds) throws RuntimeException {
        ChannelGroup groups = ImChannelGroup.CHANNELGROUP;
        for (String id : userIds) {
            Iterator<Channel> iterable = groups.iterator();
            while (iterable.hasNext()) {
                Channel next = iterable.next();
                String userId = next.attr(Constants.SessionConfig.SERVER_SESSION_ID).get();
                //这个人在线就去提醒一下有系统消息
                if (userId.equals(id)) {
                    MessageProto.Model.Builder builder = MessageProto.Model.newBuilder();
                    builder.setCmd(Constants.CmdType.GROUP_DISSOLUTION_PUSH_MESSAGE);
                    builder.setMsgtype(msgType);
                    builder.setSign(obj);
                    next.attr(Constants.SessionConfig.SERVER_SESSION_HEARBEAT).set(System.currentTimeMillis());
                    ImChannelGroup.broadcastGroup(next, builder.build());
                }
            }
        }
    }

    @Override
    public void pushMessage(MessageWrapper wrapper) throws RuntimeException {
        try {
            if (wrapper.isOffline()) {
                ImChannelGroup.broadcast(messageProxyImpl.getOffLineStateMsg(wrapper.getSessionId()));
                messageProxyImpl.updateOnlineStatusToDB(wrapper.getSessionId(), Constants.UserType.HIDE);
            } else if (wrapper.isOnline()) {
                ImChannelGroup.broadcast(messageProxyImpl.getOnLineStateMsg(wrapper.getSessionId()));
                messageProxyImpl.updateOnlineStatusToDB(wrapper.getSessionId(), Constants.UserType.ONLINE);
            } else {
                //sessionManagerImpl.send(wrapper.getSessionId(), wrapper.getBody());
                Session session = sessionManagerImpl.getSession(wrapper.getSessionId());
                /*
                 * 服务器集群时，可以在此
                 * 判断当前session是否连接于本台服务器，如果是，继续往下走，如果不是，将此消息发往当前session连接的服务器并 return
                 * if(session!=null&&!session.isLocalhost()){//判断当前session是否连接于本台服务器，如不是
                 * //发往目标服务器处理
                 * return;
                 * }
                 */
                if (session != null) {
                    boolean result = session.write(wrapper.getBody());
                    return;
                }
            }
        } catch (Exception e) {
            log.error("connector pushMessage  Exception.", e);
            throw new RuntimeException(e.getCause());
        }
    }


    @Override
    public void pushMessage(String sessionId, MessageWrapper wrapper) throws RuntimeException {
        Session session = sessionManagerImpl.getSession(sessionId);
        if (session == null) {
            throw new RuntimeException(String.format("session %s is not exist.", sessionId));
        }
        try {
            //保存到数据库
            String msgId = messageProxyImpl.saveMessageToDB(wrapper);

            MessageProto.Model model = (MessageProto.Model) wrapper.getBody();
            MessageProto.Model.Builder result = MessageProto.Model.newBuilder(model);
            result.setMsgId(msgId);
            model = MessageProto.Model.parseFrom(result.build().toByteArray());
            ///取得接收人 给接收人写入消息
            Session responseSession = sessionManagerImpl.getSession(wrapper.getReSessionId());
            if (responseSession != null && responseSession.isConnected()) {
                responseSession.getSession().attr(Constants.SessionConfig.SERVER_SESSION_HEARBEAT).set(System.currentTimeMillis());
                responseSession.write(model);
            }
            return;
        } catch (IllegalStateException e) {
            log.error("connector send occur PushException.", e);

            throw new RuntimeException(e.getCause());
        } catch (Exception e) {
            log.error("connector send occur Exception.", e);
            throw new RuntimeException(e.getCause());
        }

    }

    @Override
    public boolean validateSession(MessageWrapper wrapper) throws RuntimeException {
        try {
            return sessionManagerImpl.exist(wrapper.getSessionId());
        } catch (Exception e) {
            log.error("connector validateSession Exception!", e);
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void close(ChannelHandlerContext hander, MessageWrapper wrapper) {
        String sessionId = getChannelSessionId(hander);
        if (StringUtils.isNotBlank(sessionId)) {
            close(hander);
            log.warn("connector close channel sessionId -> " + sessionId + ", ctx -> " + hander.toString());
        }
    }

    @Override
    public void close(ChannelHandlerContext hander) {
        String sessionId = getChannelSessionId(hander);
        try {
            String nid = hander.channel().id().asShortText();
            Session session = sessionManagerImpl.getSession(sessionId);
            if (session != null) {
                sessionManagerImpl.removeSession(sessionId, nid);
                ImChannelGroup.remove(hander.channel());
                messageProxyImpl.updateOnlineStatusToDB(sessionId, Constants.UserType.OFFLINE);
                log.info("connector close sessionId -> " + sessionId + " success ");
            }
        } catch (Exception e) {
            log.error("connector close sessionId -->" + sessionId + "  Exception.", e);
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void close(String sessionId) {
        try {
            Session session = sessionManagerImpl.getSession(sessionId);
            if (session != null) {
                sessionManagerImpl.removeSession(sessionId);
                List<Channel> list = session.getSessionAll();
                for (Channel ch : list) {
                    ImChannelGroup.remove(ch);
                }
                messageProxyImpl.updateOnlineStatusToDB(sessionId, Constants.UserType.OFFLINE);
                log.info("connector close sessionId -> " + sessionId + " success ");
            }
        } catch (Exception e) {
            log.error("connector close sessionId -->" + sessionId + "  Exception.", e);
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void connect(ChannelHandlerContext ctx, MessageWrapper wrapper) {
        try {
            String sessionId = wrapper.getSessionId();
            String sessionId0 = getChannelSessionId(ctx);
            //当sessionID存在或者相等  视为同一用户重新连接
            if (StringUtils.isNotEmpty(sessionId0) || sessionId.equals(sessionId0)) {
                log.info("connector reconnect sessionId -> " + sessionId + ", ctx -> " + ctx.toString());
                pushMessage(messageProxyImpl.getReConnectionStateMsg(sessionId0));
            } else {
                log.info("connector connect sessionId -> " + sessionId + ", sessionId0 -> " + sessionId0 + ", ctx -> " + ctx.toString());
                sessionManagerImpl.createSession(wrapper, ctx);
                setChannelSessionId(ctx, sessionId);
                log.info("create channel attr sessionId " + sessionId + " successful, ctx -> " + ctx.toString());
            }
            ImUser imUser = imUserService.getOneById(sessionId);
            SysUser sysUser = sysUserService.getById(imUser.getUserId());
            List<SessionInformation> sessionInformations = sessionRegistry.getAllSessions(sysUser,false);
            if(sessionInformations.isEmpty()){
                pushSystemMessage("", Constants.SystemMsgType.LOGIN_TIME_OUT, sessionId);
            }else{
                messageProxyImpl.updateOnlineStatusToDB(sessionId, Constants.UserType.ONLINE);
            }
        } catch (Exception e) {
            log.error("connector connect  Exception.", e);
        }
    }

    @Override
    public boolean exist(String sessionId) throws Exception {
        return sessionManagerImpl.exist(sessionId);
    }

    @Override
    public String getChannelSessionId(ChannelHandlerContext ctx) {
        return ctx.channel().attr(Constants.SessionConfig.SERVER_SESSION_ID).get();
    }

    private void setChannelSessionId(ChannelHandlerContext ctx, String sessionId) {
        ctx.channel().attr(Constants.SessionConfig.SERVER_SESSION_ID).set(sessionId);
    }


}
